package com.example.jetpacktest.coroutine

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.flow.internal.SendingCollector
import org.junit.Test
import java.lang.Exception
import kotlin.system.measureTimeMillis

class Test003 {

    @Test
    fun test001(){
        val s: Sequence<Int> = sequence {
            for(i in 1..3){
                Thread.sleep(1000)
                yield(i)
            }
        }
        s.forEach {
            println(it)
        }
    }


    @Test
    fun test002(){
        runBlocking {
            launch {
                for(i in 1..3){
                    println("not block the thread...$i")
                    delay(2000)
                }
            }

            f001().collect{ value -> println(value) }
        }
    }


    private fun f001(): Flow<Int>{
        return flow {
            for(i in 1..3){
                delay(1000)
                emit(i)
            }
        }
    }


    @Test
    fun test003(){
        runBlocking {
            val flow1 = f002()
            println("call collect")
            flow1.collect {
                println("first...$it")
            }
            println("call collect again")
            flow1.collect {
                println("again....$it")
            }
        }
    }

    private fun f002(): Flow<Int>{
        return flow {
            println("flow start")
            for(i in 1..3){
                delay(10000)
                emit(i)
            }
        }
    }

    @Test
    fun test004(){
        runBlocking {
            val res = withTimeoutOrNull(1500){
                try{
                    f004().collect {
                        println("receive......$it")
                    }
                }catch (ex: Exception){
                    println("$ex")
                }

                10
            }
        }
    }


    private fun f004(): Flow<Int>{
        return flow {
            for(i in 1..3){
                delay(1000)
                emit(i)
                println("emit....$i")
            }

        }
    }


    @Test
    fun test005(){
        runBlocking {
            listOf<Int>(1, 2, 3).asFlow().collect { println(it) }
        }
    }


    /**
     * 执行转换
     */
    suspend fun f006(x: Int): String{
        delay(2000)
        return "response...$x"
    }

    @Test
    fun test006() {
        runBlocking {
            (1..3).asFlow()
                .map {
                    f006(it)
                }.collect {
                    println(it)
                }
        }
    }

    @Test
    fun test007(){
        runBlocking {
            (1..3).asFlow().transform {
                emit("emit before....$it")
                emit(f006(it))
            }.collect {
                println(it)
            }
        }
    }


    fun f008(): Flow<Int>{
        return flow<Int> {
            try{
                emit(1)
                emit(2)
                println("this line will not executed")
                emit(3)
            }finally {
                println("except...finally")
            }

        }
    }

    @Test
    fun test008(){
        runBlocking {
            f008().take(4).collect {
                println(it)
            }

            f008().take(2).collect {
                println(it)
            }
        }
    }

    @Test
    fun test009(){
        runBlocking {
            val x = (1..5).asFlow().map {
                it * it
            }.first()
            println(x)
        }
    }

    @Test
    fun test010(){
        runBlocking {
            val x = (1..5).asFlow().reduce{a, b ->
                println("a = $a, b = $b")
                a + b
            }
            println("x = $x")
        }
    }


    private fun f011(): Flow<Int>{
        return flow {
            println("flow's thread = ${Thread.currentThread()}")
            for(i in 1..3){
                emit(i)
            }
        }
    }

    @Test
    fun test011(){
        runBlocking {
            launch(Dispatchers.IO) {
                f011().collect {
                    println("$it")
                }
            }

            launch {
                f011().collect {
                    println("$it")
                }
            }
        }
    }


    private fun f012(): Flow<Int>{
        return flow {
            println("send thread = ${Thread.currentThread()}")
            for(i in 1..5){
                delay(1000)
                emit(i)
            }
        }.flowOn(Dispatchers.IO)
    }

    @Test
    fun test012(){
        runBlocking {
            launch{
                f012().collect {
                    println("collect thread = ${Thread.currentThread()}")
                    println("$it")
                }
            }
        }
    }


    private fun f013(): Flow<Int>{
        return flow {
            for(i in 1..3){
                delay(1000)
                emit(i)
            }
        }
    }

    @Test
    fun test013(){
        runBlocking {
            val time = measureTimeMillis {
                f013()
                    .buffer()
                    .collect{
                        delay(10000)
                        println(it)
                }
            }
            println(time)

        }
    }


    private fun f014(): Flow<Int>{
        return flow<Int> {
            for(i in 1..11){
                delay(100)
                emit(i)
            }
        }
    }


    @Test
    fun test014(){
        runBlocking {
            val t = measureTimeMillis {
                f014().conflate().collect {
                    delay(300)
                    println("$it")
                }
            }
            println("$t")
        }
    }


    @Test
    fun test015(){
        runBlocking {
            val t = measureTimeMillis {
                f014().collectLatest {
                    println("start...$it")
                    delay(300)
                    println("end...$it")
                }
            }
            println("time = $t")
        }
    }


    @Test
    fun test016(){
        runBlocking {
            val flow1 = (1..3).asFlow().onEach { delay(300) }
            val flow2 = flowOf("a", "b", "c").onEach { delay(400) }
            val startTime = System.currentTimeMillis()
            flow1.zip(flow2){a, b ->
                "$a--->$b"
            }.collect {
                println("$it.....${System.currentTimeMillis() - startTime}")
            }
        }
    }


    @Test
    fun test017(){
        runBlocking {
            val flow1 = (1..6).asFlow().onEach { delay(300) }
            val flow2 = flowOf("a", "b", "c").onEach { delay(400) }
            val startTime = System.currentTimeMillis()
            flow1.combine(flow2){a, b ->
                "$a--->$b"
            }.collect {
                println("$it......${System.currentTimeMillis() - startTime}")
            }
        }
    }


    @Test
    fun test018(){
        runBlocking {
            val fOut: Flow<Flow<String>> = (1..3).asFlow().map {
                flow<String>{
                    emit("$it....")
                }
            }

            fOut.collect { fIn ->
                fIn.collect{
                    println(it)
                }
            }
        }
    }


    private fun requestFlow(x: Int): Flow<String>{
        return flow {
            emit("$x......first")
            delay(500)
            emit("$x......second")
        }
    }


    @Test
    fun test019() {
        runBlocking {
            val startTime = System.currentTimeMillis()
            (1..3).asFlow().onEach { delay(100) }
                .flatMapMerge {
                    requestFlow(it)
                }.collect {
                    println("$it.....${System.currentTimeMillis() - startTime}")
                }
        }
    }


    private fun f020(): Flow<Int>{
        return flow<Int>{
            for(i in 1..3){
                println("emit.....$i")
                emit(i)
            }
        }
    }

    @Test
    fun test020(){
        runBlocking {
            try{
               f020().collect {
                   check(it <= 1){
                       "值大于1"
                   }
                   println(it)
               }
            }catch (ex: Exception){
                println("ex = $ex")
            }
        }
    }


    private fun f021(): Flow<Int>{
        return flow {
            for(i in 1..3){
                check(i <= 2){
                    "i大于1"
                }
                emit(i)
            }
        }
    }

    @Test
    fun test021(){
        runBlocking {
            f021().catch { ex ->
                println("ex = $ex")
            }.collect {
                println(it)
            }
        }
    }

}